#include "tcpclient.h"

#include "sockutil.h"
#include "timingwheel.h"
#include "connection.h"
#include "timer.h"

#include "connector.inl"

#define maxRecvTimeout  30
#define maxConnTimeout  5
#define maxBufSize      8192
#define maxReconns      3

namespace tnet
{
    const char* tcp_strerror(int code) 
    {
        switch (code) {
        TCPCLIENT_ERRNO_MAP(TCP_STRERROR)
        default:  return "unkown other error";
        }
    }

    static void int2bytes(int value,unsigned char* bytes,int len)
    {
        for (int i = 0; i < len; i++)
            bytes[len - i - 1] = (unsigned char)((value >> 8 * i));
        bytes[len] = 0x00;
        return;
    }

    static int bytes2int(unsigned char* bytes,int len)
    {
        int sum = 0,end=len;
        for (int i = 0; i < end; i++)
        {
            int n = bytes[i] & 0xff;
            n <<= (--len) * 8;
            sum += n;
        }
        return sum;
    }

    static void dummyCallback(const ClientConnectionPtr_t&, const int,  const string&)
    {
    }

    TcpParser::TcpParser()
        : m_length(0)
        , m_first(true)
        , m_readlen(0)
    {
    }

    int TcpParser::parse(const char* buffer,const uint16_t count)
    {
        if( m_length == 0)
        {
            m_buffer = string(buffer,count);
            onMessageComplete();
        }
        else
        {
            if(m_first)
            {
                m_buffer.clear();
                char header_buffer[128]={0};
                strncpy(header_buffer,buffer,m_length);
                m_readlen = std::stoi(header_buffer);
                m_buffer.append(buffer+m_length,count-m_length);
                if(m_readlen >= count-m_length)
                {
                    onMessageComplete();
                    return 0;     
                }
                m_first = false;
            }
            else
            {
                m_buffer.append(buffer,count);
                if(m_buffer.size() >= m_readlen)
                {
                    onMessageComplete();
                    m_first = true;
                }
            }
        }
        return 0;
    }    

	ClientConnection::ClientConnection()
		: Connector<ClientConnection>()
	{
		 m_callback = std::bind(&dummyCallback, _1, _2, _3);
	}	
	
	ClientConnection::~ClientConnection()
	{
	}
	
	void ClientConnection::clearCallback()
	{
		m_callback = std::bind(&dummyCallback, _1, _2, _3);
	}
	
	void ClientConnection::handleRead(const char* buffer, size_t count)
	{
        parse(buffer,count);
	}

    int ClientConnection::onMessageComplete()
    {
        m_callback(shared_from_this(),(const int)ClientErrorCode::successed,m_buffer);
        return 0;
    }
	
	void ClientConnection::handleConnectTimeout(const void* )
	{
		ClientConnection::m_callback(shared_from_this(),(const int)ClientErrorCode::connect_timeout,string());
	}

	void ClientConnection::handleReadTimeout(const void* )
	{
		ClientConnection::m_callback(shared_from_this(),(const int)ClientErrorCode::read_timeout,string());
	}
	
	void ClientConnection::handleError(const void*)
	{
		m_callback(shared_from_this(),(const int)ClientErrorCode::handle_error,string());
	}   

    TcpClient::TcpClient(uint16_t maxClients)
        : m_maxClients(maxClients)
        , m_isIPV6(false)
        , m_port(0)
        , m_maxReadTimeout(30)
        , m_length(0)
    {
        m_loop = new IOLoop();
    }  

    TcpClient::TcpClient(const string& host,uint16_t port, uint16_t maxClients)
        : m_maxClients(maxClients)
        , m_isIPV6(false)
        , m_host(host)
        , m_port(port)
        , m_maxReadTimeout(30)
        , m_length(0)
    {
        m_loop = new IOLoop();
    }   

    TcpClient::~TcpClient()
    {
        disConnect();
        if(m_loop)
        {
            delete m_loop;
            m_loop=nullptr;
        }
    }  

    void TcpClient::onResponse(const ClientConnectionPtr_t& conn, const int errorCode, const string& responseData,int* response_code,string* response_data)
    { 
        //add refer
        ClientConnectionPtr_t c = conn->shared_from_this();
        if(errorCode == 0)
        {
            pushConn(conn);
        }
        m_loop->stop();
        if( responseData.size() > 0 )
        {
            *response_data = responseData;
        }
        *response_code= errorCode;
    }	      

    void TcpClient::send(const std::string& host, const uint16_t port,const string& request_data, int& response_code,string& response_data)
    {
        m_host = host;
        m_port = port;
        return send(request_data,response_code,response_data);
    }  

    void TcpClient::send(const string& request_data, int& response_code,string& response_data)
    {
        if(m_host.empty())
        {
            response_code = static_cast<int>(ClientErrorCode::addr_error);
            return;
        }

        if(m_port == 0)
        {
            response_code =  static_cast<int>(ClientErrorCode::port_error);
            return;            
        }

        string new_request_data = std::move(request_data);
        if( m_length > 0 )
        {
            char length_buf[256]={0};
            sprintf(length_buf,"%0*d",m_length,new_request_data.size());
            new_request_data.insert(0,length_buf);
        }

        //now we only support ip
        Address addr(m_host,m_port,m_isIPV6);
        ClientConnectionPtr_t conn = popConn(addr.ip());
        if(conn)
        {
            conn->setCallback(std::bind(&TcpClient::onResponse, shared_from_this(), _1, _2, _3,&response_code, &response_data));
            conn->send(new_request_data);
        }
        else
        {
            conn = std::make_shared<ClientConnection>();
            conn->setReadTimeout(m_maxReadTimeout);
            conn->setLength(m_length);
            if( conn->connect(m_loop, addr, std::bind(&TcpClient::onConnect, shared_from_this(), 
                _1, _2, new_request_data, &response_code,&response_data), m_device) < 0)
            {
                response_code =  static_cast<int>(ClientErrorCode::connect_error);
                return;
            }
        }
        m_loop->start();
    }

    void TcpClient::onConnect(const ClientConnectionPtr_t& conn, bool connected, const std::string& requestData,int* errorCode,string* response_data)
    {
        if(!connected) 
        {
            LOG_ERROR("TcpClient connect error");
            *errorCode = -1;
            m_loop->stop();
            return;
        } 
        string data = std::move(requestData);
        conn->setCallback(std::bind(&TcpClient::onResponse, shared_from_this(), _1, _2, _3,errorCode,response_data));
        conn->send(data);			
    }  

    void TcpClient::disConnect()
    {
        IpConn_t::iterator iter = m_conns.begin();
        while(iter != m_conns.end())
        {
            ClientConnectionPtr_t conn = iter->second.lock();
            if(conn)
            {
                conn->shutDown();    
            }
            ++iter;    
        } 
        m_conns.clear();        
    }   

    void TcpClient::pushConn(const ClientConnectionPtr_t& conn)
    {
        conn->clearCallback();
         
        ConnectionPtr_t c = conn->lockConn();
        if(!c)
        {
            return;
        }

        if(m_conns.size() >= m_maxClients)
        {
            conn->shutDown();
            return;
        }

        Address addr(0);
        if(SockUtil::getRemoteAddr(c->getSockFd(), addr,m_isIPV6) != 0)
        {
            conn->shutDown();
            return;    
        }
        m_conns.insert(make_pair(addr.ip(), conn));       
    }
    
    ClientConnectionPtr_t TcpClient::popConn(uint32_t ip)
    {
        while(true)
        {
            IpConn_t::iterator iter = m_conns.find(ip);
            if(iter == m_conns.end())
            {
                return ClientConnectionPtr_t();        
            }
            else
            {
                WeakClientConnectionPtr_t conn = iter->second;
                m_conns.erase(iter);
                ClientConnectionPtr_t c = conn.lock(); 
                if(c && c->lockConn())
                {
                    return c;    
                }   
            }
        }       
    }     

    SocketClient::SocketClient()
     : m_sockFd(-1)
     , m_maxReadTimeout(maxRecvTimeout)
     , m_maxConnTimeout(maxConnTimeout)
     , m_length(0)
     , m_isBlock(true)
     , m_isIpv6(false)
     , m_isReconn(false)
     , m_isKeepalive(false)
    {

    };

    SocketClient::~SocketClient()
    {
        closeClient();
    }

    int SocketClient::closeClient()
    {
        if(m_sockFd)
        {
            close(m_sockFd);
            m_sockFd=-1;
        }        
        return 0;
    }

    void SocketClient::updateActiveTime()
    {
        struct timespec t;
        clock_gettime(CLOCK_MONOTONIC, &t);
        m_lastActiveTime = t.tv_sec;
    }    

    int SocketClient::connect(const Address& address)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            m_isIpv6 = address.isIPV6();
            m_sockFd = SockUtil::create(m_isBlock,m_isIpv6);
            if(m_sockFd < 0)
            {
                return TCP_RETCODE::TCP_CREATE_ERROR;
            }
            if(m_isKeepalive)
            {
                SockUtil::setKeepAlive(m_sockFd,m_isKeepalive);
            }
            if( (iRet = SockUtil::connect(m_sockFd,address)) != 0)
            {
                if(errno == EINPROGRESS)
                {
                    iRet = SockUtil::isWriteAble(m_sockFd,m_maxConnTimeout);
                    if(iRet == 0)
                        return TCP_RETCODE::TCP_CONNECT_TIMEOUT;
                    else if(iRet < 0)
                        return TCP_RETCODE::TCP_CONNECT_ERROR;
                }
                else 
                {
                    return TCP_RETCODE::TCP_CONNECT_ERROR;         
                }
            }
            strcpy(m_addr.ipaddr.ip,address.ipstr().c_str());
            m_addr.ipaddr.port = address.port();
            updateActiveTime();
        }        
        return TCP_RETCODE::TCP_SUCESSED;
    }

    int SocketClient::send(const string& sndBuf,string& rcvBuf)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            return TCP_ARGS_ERROR;
        }
        else
        {
            if( SockUtil::isClosed(m_sockFd) )
            {
                closeClient();
                if(!m_isReconn)
                {
                    return TCP_RETCODE::TCP_PEER_CLOSE;
                }
                iRet = connect(Address(m_addr.ipaddr.ip,m_addr.ipaddr.port));
                if( iRet != TCP_RETCODE::TCP_SUCESSED)    
                {
                    return iRet;
                }
            }
        }  
        return sendRecv(sndBuf,rcvBuf);
    };    

    int SocketClient::send(const Address& address,const string& sndBuf,string& rcvBuf)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            if( (iRet=connect(address)) != TCP_RETCODE::TCP_SUCESSED)
            {
                return iRet;
            }     
        }
        return send(sndBuf,rcvBuf);
    }

    int SocketClient::sendRecv(const string& sndBuf,string& rcvBuf)
    {
        int iRet = 0;
        if( (iRet = send(sndBuf)) != TCP_RETCODE::TCP_SUCESSED )
        {
            return TCP_RETCODE::TCP_SEND_ERROR;
        }
        return recv(rcvBuf);
    } 

    int SocketClient::send(const string& sndBuf)
    {
        string ssndBuf(sndBuf);

        if(m_sockFd < 0)
        {
            return TCP_ARGS_ERROR;
        }

        char sndHeader[256]={0};
        uint32_t nSndMesgLen = sndBuf.length();
        if(m_length > 0)
        {
            if(m_length == 2)
                int2bytes(nSndMesgLen,(unsigned char*)sndHeader,m_length);
            else
                sprintf(sndHeader,"%0*d",m_length,nSndMesgLen); 	
            ssndBuf.insert(0,sndHeader);
        }
 
        if( SockUtil::send(m_sockFd,ssndBuf) != ssndBuf.length())
        {
            return TCP_RETCODE::TCP_SEND_ERROR;
        }
        return TCP_RETCODE::TCP_SUCESSED;
    }

    int SocketClient::recv(string& rcvBuf)
    {
        int iRet = SockUtil::isReadAble(m_sockFd,m_maxReadTimeout);
        if(iRet < 0)
            return TCP_RETCODE::TCP_OTHER_ERROR;
        else if(iRet == 0)
            return TCP_RETCODE::TCP_RECV_TIMEOUT;

        int option = 0;
        uint32_t nRcvMesgLen = maxBufSize;
        if(m_length > 0)
        {
            string rcvHeader;
            iRet = SockUtil::recv(m_sockFd,rcvHeader,m_length);
            if(  iRet < 0)
                return TCP_RETCODE::TCP_RECV_ERROR;
            else if(iRet == 0)
            {
                closeClient();
                return TCP_RETCODE::TCP_PEER_CLOSE;
            }
            if(m_length == 2)
                nRcvMesgLen = bytes2int((unsigned char*)rcvHeader.c_str(),m_length);
            else
                nRcvMesgLen = std::stoi(rcvHeader);           
        }
        if(nRcvMesgLen == 0)
        {
            return TCP_RETCODE::TCP_RCVDATA_INVALID;
        }
        iRet = SockUtil::recv(m_sockFd,rcvBuf,nRcvMesgLen,option);
        if(  iRet < 0)
            return TCP_RETCODE::TCP_RECV_ERROR;
        else if(iRet == 0)
        {
            closeClient();
            return TCP_RETCODE::TCP_PEER_CLOSE;
        }
        updateActiveTime();
        return TCP_RETCODE::TCP_SUCESSED;
    }     

    int UnixSocketClient::connect(const string&  unPath)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            m_sockFd = SockUtil::createUDS(m_isBlock);
            if(m_sockFd < 0)
            {
                return TCP_RETCODE::TCP_CREATE_ERROR;
            }
            if(  SockUtil::connectUDS(m_sockFd,unPath) < 0)
            {
                if(errno == EINPROGRESS)
                {
                    iRet = SockUtil::isWriteAble(m_sockFd,m_maxConnTimeout);
                    if(iRet == 0)
                        return TCP_RETCODE::TCP_CONNECT_TIMEOUT;
                    else if(iRet < 0)
                        return TCP_RETCODE::TCP_CONNECT_ERROR;
                }
                else 
                {
                    return TCP_RETCODE::TCP_CONNECT_ERROR;         
                }
            }
            strcpy(m_addr.un_path,unPath.c_str());
        }       
        return TCP_RETCODE::TCP_SUCESSED;
    }   

    int UnixSocketClient::send(const string& unPath,const string& sndBuf,string& rcvBuf)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            if( (iRet=connect(unPath)) != TCP_RETCODE::TCP_SUCESSED)
            {
                return iRet;
            }          
        }
        return send(sndBuf,rcvBuf);
    }   

    int UnixSocketClient::send(const string& sndBuf,string& rcvBuf)
    {
        int iRet = 0;
        if(m_sockFd < 0)
        {
            return TCP_ARGS_ERROR;
        }
        else
        {
            if( SockUtil::isClosed(m_sockFd) )
            {
                closeClient();
                if(!m_isReconn)
                {
                    return TCP_RETCODE::TCP_PEER_CLOSE;
                }
                iRet = connect(m_addr.un_path);
                if( iRet != TCP_RETCODE::TCP_SUCESSED)    
                {
                    return iRet;
                }                
            }
        } 
        return sendRecv(sndBuf,rcvBuf);     
    }

}  // end namespace tnet;

